﻿#include <boost/beast/core.hpp>
#include <boost/beast/websocket.hpp>
#include <boost/asio/connect.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/buffer.hpp>
#include <boost/filesystem.hpp>

#include <cstdlib>
#include <functional>
#include <iostream>
#include <fstream>
#include <sstream>
#include <memory>
#include <string>
#include <chrono>
#include <QDebug>
#include <QThread>
#include "async_client.h"

using tcp = boost::asio::ip::tcp;  // from <boost/asio/ip.tcp.hpp>
namespace websocket = boost::beast::websocket;  // from <boost/beast/websocket.hpp>

// Report a failure
void fail(boost::system::error_code ec, char const* what)
{
  std::cerr << what << ": " << ec.message() << "\n";
}

// Sends a WebSocket message and prints the response
class session : public std::enable_shared_from_this<session>
{
  tcp::resolver resolver_;
  websocket::stream<tcp::socket> ws_;
  boost::beast::multi_buffer buffer_;
  boost::beast::flat_buffer rxBuffer_;
  std::string host_;
  std::string text_;
  //数据缓冲队列
  std::queue<std::string> recvcache;
  std::queue<std::string> sendcache;
  std::mutex rlock;
  std::mutex slock;

public:
  // Resolver and socket require an io_context
  explicit session(boost::asio::io_context& ioc)
    : resolver_(ioc), ws_(ioc)
  {}

  // Start the asynchronous operation
  void run(char const* host, char const* port, char const* text)
  {
    // Save these for later
    host_ = host;
    text_ = text;

    // Look up the domain name
    resolver_.async_resolve(host, port, std::bind(
      &session::on_resolve, shared_from_this(), std::placeholders::_1, std::placeholders::_2));
  }

  void on_resolve(boost::system::error_code ec, tcp::resolver::results_type results)
  {
    if (ec) return fail(ec, "resolve");

    // Make the connection on the IP address we get from a lookup
    boost::asio::async_connect(ws_.next_layer(), results.begin(), results.end(),
      std::bind(&session::on_connect, shared_from_this(), std::placeholders::_1));
  }

  void web_connect(char const* host, char const* port)
  {
      host_ = host;
      auto const results = resolver_.resolve(host, port);
      boost::asio::async_connect(ws_.next_layer(), results.begin(), results.end(),
        std::bind(&session::on_connect, shared_from_this(), std::placeholders::_1));
  }
  void on_connect(boost::system::error_code ec)
  {
    if (ec) return fail(ec, "connect");

    // Perform the websocket handshake
    ws_.async_handshake(host_, "/",
      std::bind(&session::on_handshake, shared_from_this(), std::placeholders::_1));
  }

  void on_handshake(boost::system::error_code ec)
  {
    if (ec) return fail(ec, "handshake");
    ws_.read_message_max(1 << 27);//64MB

    //// Read a message into our buffer
    //ws_.async_read(rxBuffer_, std::bind(
    //  &session::on_read, shared_from_this(), std::placeholders::_1, std::placeholders::_2));
    ws_.async_read(rxBuffer_, boost::beast::bind_front_handler(
        &session::on_read, shared_from_this()
    ));
//    ws_.async_write(boost::asio::buffer(text_), std::bind(
//      &session::on_write, shared_from_this(), std::placeholders::_1, std::placeholders::_2));
  }

  void sendAsync(std::string sdata){
      ws_.binary(true);
      text_ = sdata;
          ws_.async_write(boost::asio::buffer(text_), std::bind(
            &session::on_write, shared_from_this(), std::placeholders::_1, std::placeholders::_2));
  }

  void on_write(boost::system::error_code ec, std::size_t bytes_transferred)
  {
    boost::ignore_unused(bytes_transferred);

    if (ec) return fail(ec, "write");

    //// Read a message into our buffer
    //ws_.async_read(buffer_, std::bind(
    //  &session::on_read, shared_from_this(), std::placeholders::_1, std::placeholders::_2));
  }

  void on_read(boost::system::error_code ec, std::size_t bytes_transferred)
  {
    boost::ignore_unused(bytes_transferred);
    std::cout << ec << std::endl;
    qDebug() << bytes_transferred;
    if (ec) return fail(ec, "read");
    if(ws_.text()){
         qDebug() << Q_FUNC_INFO;
         std::cout << bytes_transferred<<std::endl;
         std::cout << "hello world"<< std::endl;
        char *tmp = new char[bytes_transferred];
        //for(auto it : buffer_.data())
        //{
        //    memcpy(tmp, it.data(), it.size());
        //}
        memcpy(tmp, rxBuffer_.data().data(), rxBuffer_.data().size());
        std::cout << tmp << std::endl;
        delete[] tmp;
    }
    else {
        std::string _rcv_str;
        _rcv_str.assign((char *)rxBuffer_.data().data(), rxBuffer_.data().size());
        std::unique_lock<std::mutex> lock_reg(rlock);
        recvcache.push(_rcv_str);
        int l1 = rxBuffer_.size() ;
        int l2 = rxBuffer_.data().size() ;
        std::cout << l1<< std::endl;
        std::cout << l2<< std::endl;
        std::cout << _rcv_str.size() << std::endl;
    }
//    std::cout << boost::beast::buffers(buffer_.data()) << std::endl;
    std::cout << rxBuffer_.size() << std::endl;
    rxBuffer_.consume(rxBuffer_.size());
    ws_.async_read(rxBuffer_, std::bind(
      &session::on_read, shared_from_this(), std::placeholders::_1, std::placeholders::_2));


//    on_write(ec, bytes_transferred);

//    // Close the WebSocket connection
//    ws_.async_close(websocket::close_code::normal, std::bind(
//      &session::on_close, shared_from_this(), std::placeholders::_1));
  }

  void on_close(boost::system::error_code ec)
  {
    if (ec) return fail(ec, "close");

    // If we get here then the connection is closed gracefully

    // The buffers() function helps print a ConstBufferSequence
//    std::cout << boost::beast::buffers(buffer_.data()) << std::endl;
  }
  std::string get_recv(void) {
      std::string _str;
      if (!recvcache.empty()) {
          std::unique_lock<std::mutex> lock_reg(rlock);
          _str = recvcache.front();
          recvcache.pop();
      }
      return _str;
  }
};
static std::shared_ptr<session> session_sptr;
AsyncProcess::AsyncProcess(QObject *parent) : QObject(parent)
{

}

void AsyncProcess::startWebSocket()
{
    qDebug() << Q_FUNC_INFO;
//    auto const host = "127.0.0.1";
//    auto const port = "42002";
    auto const text = "hello";
    boost::asio::io_context ioc;
    std::make_shared<session>(ioc)->run(host.c_str(), port.c_str(), text);
    // Run the I/O service. The call will return when the socket is closed.
    ioc.run();
}
void AsyncProcess::connectWebSocketServer()
{
    boost::asio::io_context ioc;
    session_sptr =   std::make_shared<session>(ioc);
    session_sptr->web_connect(host.c_str(), port.c_str());
    ioc.run();
    session_sptr.reset();
}
void AsyncProcess::sendData(QString  qfile_name)
{
    std::string file_name = qfile_name.toStdString();
    std::string gen;

    if (boost::filesystem::is_regular_file(file_name)){
        std::ifstream fp;
        fp.open(file_name, std::ios::in | std::ios::binary);
        int length = boost::filesystem::file_size(file_name);
        qDebug()<< QString::fromStdString(file_name) << "   [ " << boost::filesystem::file_size(file_name) << " ]\n";
        char *pdata = new char[length];
        fp.read(pdata, length);
        fp.close();
        gen.assign(pdata, length);
        free(pdata);
    }
    session_sptr->sendAsync(gen);
}
void AsyncProcess::getRecvData(cv::Mat & r_Img) {
   
    std::string res = session_sptr->get_recv();
    r_Img.create(2704, 2720, CV_16UC1);
    if (res.size() == r_Img.total() * 2)
        memcpy(r_Img.data, res.c_str(), r_Img.total() * 2);
}
//void AsyncProcess::sendData()
//{
//    std::string file_name = "G:/20_data/raw/20210808/B4-Frame00007715.raw";
//    std::ifstream fp;
//    fp.open(file_name, std::ios::in | std::ios::binary);

//    if (boost::filesystem::is_regular_file(file_name)){
//        std::cout << file_name << "   [ " << boost::filesystem::file_size(file_name) << " ]\n";
//    }
//    session_sptr->sendAsync();
//}
void AsyncProcess::stopWebSocket()
{
    qDebug() << Q_FUNC_INFO;
}

